package org.gbif.metrics.cli;

import org.gbif.api.model.occurrence.Occurrence;
import org.gbif.common.messaging.api.MessageCallback;
import org.gbif.common.messaging.api.OccurrencePersistedMessage;
import org.gbif.metrics.cube.HBaseCubes;

import java.util.concurrent.TimeUnit;

import com.urbanairship.datacube.AsyncException;
import com.urbanairship.datacube.Batch;
import com.urbanairship.datacube.Op;
import com.yammer.metrics.Metrics;
import com.yammer.metrics.core.Counter;
import com.yammer.metrics.core.Timer;
import com.yammer.metrics.core.TimerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * Base class to simplify writing message callbacks that mutate cubes by providing the message handling and the timing
 * services.
 * 
 * @param <T> The cube type
 */
abstract class CubeMutatorCallback<T extends Op> implements MessageCallback<OccurrencePersistedMessage> {

  private static final Logger LOG = LoggerFactory.getLogger(CubeMutatorCallback.class);
  private final Counter messageCount = Metrics.newCounter(getClass(), "messageCount");
  private final Counter newOccurrencesCount = Metrics.newCounter(getClass(), "newOccurrencesCount");
  private final Counter updatedOccurrencesCount = Metrics.newCounter(getClass(), "updatedOccurrencesCount");
  private final Timer writeTimer = Metrics.newTimer(getClass(), "cubeWrites", TimeUnit.MILLISECONDS, TimeUnit.SECONDS);

  protected abstract HBaseCubes<T> getCube();

  protected abstract Batch<T> getNewMutations(Occurrence occurrence);

  protected abstract Batch<T> getUpdateMutations(Occurrence oldOccurrence, Occurrence newOccurrence);

  // TODO: Maybe consider retry upon exception
  @Override
  public void handleMessage(OccurrencePersistedMessage message) {
    LOG.debug("Handling {} occurrence", message.getStatus());
    messageCount.inc();

    Batch<T> update = null;
    switch (message.getStatus()) {
      case NEW:
        newOccurrencesCount.inc();
        update = getNewMutations(message.getNewOccurrence());
        break;
      case UPDATED:
        updatedOccurrencesCount.inc();
        update = getUpdateMutations(message.getOldOccurrence(), message.getNewOccurrence());
        break;
      case UNCHANGED:
      default:
        return;
    }

    TimerContext context = writeTimer.time();
    try {
      if (update.getMap() != null && !update.getMap().isEmpty()) {
        LOG
          .debug("Mutations for {} occurrence produced {} cube mutations", message.getStatus(), update.getMap().size());
        getCube().write(update);
      } else {
        LOG.debug("No mutations for {} occurrence produced. Old[{}] New[], ", message.getStatus(),
          message.getOldOccurrence(), message.getNewOccurrence());
      }
    } catch (AsyncException e) {
      LOG.error("Exception while updating cube for [{}] Occurrence [{}]", message.getStatus(), e);
    } catch (InterruptedException e) {
      LOG.error("Interrupted while updating cube for [{}] Occurrence [{}]", message.getStatus(), e);
    } finally {
      context.stop();
    }
  }
}
